package com.andox.fn.inf.rpc.rest.protocol.registry.zookeeper;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.andox.fn.inf.rpc.rest.common.Constants;
import com.andox.fn.inf.rpc.rest.config.InterfaceConfig;
import com.andox.fn.inf.rpc.rest.config.ReferenceConfig;
import com.andox.fn.inf.rpc.rest.config.ServiceConfig;
import com.andox.fn.inf.rpc.rest.protocol.registry.RegistryImpl;
import com.andox.fn.inf.rpc.rest.protocol.registry.RegistryService.RegistryNotify;
import com.andox.fn.inf.rpc.rest.util.ReflectUtil;
import com.andox.fn.inf.rpc.rest.util.StringUtil;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.netflix.curator.RetryPolicy;
import com.netflix.curator.RetrySleeper;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.CuratorFrameworkFactory;
import com.netflix.curator.framework.api.CuratorWatcher;


/**   
 * @ClassName:   ZookeeperRegistryImpl   
 * @author:      Andox
 * @date:        2019年8月16日 下午1:53:05   
 *     						inf-rpc-translator
 *     					/			|		\
 *     				ServiceA	ServiceB	ServiceC
 *     			/     |       \
 *     	consumer     provider	...其他扩展
 *     /     |       /    \
 *    SC(E) Sb(E) NodeM(E) NodeN(E)			
 *     
 *    E : 临时节点
 *     
 */
public class ZookeeperRegistryImpl implements RegistryImpl {
	private final static Logger logger = LoggerFactory.getLogger(ZookeeperRegistryImpl.class);
	
	private final static String SPLIT = "/";
	private final static String PROVIDER_PATH_FORMAT = "/%s/provider";
	private final static String CONSUME_PATH_FORMAT = "/%s/consumer";
	
	private String connection;
	private CuratorFramework curatorFramework;
	
	@Override
	public void start() {
		RetryPolicy retryPolicy = new RetryPolicy() {
			@Override
			public boolean allowRetry(int retryCount, long elapsedTimeMs, RetrySleeper sleeper) {
				if (retryCount > 3)
					try {
						logger.error("Curator frame connecting zookeeper [{}], elapsedTimeInMilliseconds: {}, retry index: {}", 
								connection, elapsedTimeMs, retryCount);
						sleeper.sleepFor(1000, TimeUnit.MILLISECONDS);
					} catch (InterruptedException ignore) {
					}
					
				return true;
			}
		};
		
		this.curatorFramework = CuratorFrameworkFactory.builder()
				.connectString(connection)
				.connectionTimeoutMs(3000)
				.sessionTimeoutMs(1000)
				.retryPolicy(retryPolicy)
				.namespace(Constants.ZOOKEEPER_REGISTRY_NAMESPACE)
				.build();
		this.curatorFramework.start();
	}

	@Override
	public void close() {
		if (this.curatorFramework != null)
			this.curatorFramework.close();
	}

	@Override
	public void setConnection(String connection) {
		this.connection = connection;
	}
	
	public boolean checkExist(String path) {
		try {
			Stat stat = this.curatorFramework.checkExists().forPath(path);
			return stat != null;
		} catch (Exception e) {
			throw new ZookeeperException(e);
		}
	}
	
	public void createPersistPath(String path) {
		try {
			if ("/".equals(path) || this.checkExist(path))
				return;
			else {
				// 创建父目录
				int i = path.lastIndexOf("/");
				Preconditions.checkArgument(i >= 0);
				String parent = path.substring(0, i);

				this.createPersistPath(parent);
			}
			this.curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath(path, null);
		} catch (Exception e) {
			throw new ZookeeperException(e);
		}
	}
	
	public void createEphemeralPath(String path) {
		try {
			if ("/".equals(path) || this.checkExist(path))
				return;
			else {
				// 创建父目录
				int i = path.lastIndexOf("/");
				Preconditions.checkArgument(i >= 0);
				String parent = path.substring(0, i);

				this.createPersistPath(parent);
			}
			this.curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath(path, null);
		} catch (Exception e) {
			throw new ZookeeperException(e);
		}
	}

	private String registerData2String(String basePath, InterfaceConfig rd) {
		StringBuilder sb = new StringBuilder(rd.getInterfaceClass().getName());
		
		Map<String, Object> properties = new HashMap<>();
		ReflectUtil.findJavaProperties(rd, InterfaceConfig.class, "", properties);
		String s = StringUtil.properties2UrlString(properties).toString();
		sb.append("?").append(s);
		
		return basePath + SPLIT + sb.toString();
	}
	
	@Override
	public void register(ServiceConfig serviceConfig) {
		String path = this.registerData2String(String.format(PROVIDER_PATH_FORMAT, serviceConfig.getInterfaceClass().getName()), serviceConfig);
		this.createEphemeralPath(path);
	}

	@Override
	public void unregister(ServiceConfig serviceConfig) {
		// 好吧, 暂时服务器关闭, 自动释放
	}

	/**
	 * 监控provider child, 刷新客户端Invoker, 集群, 及其他
	 * @param path
	 * @param rn
	 * @return
	 */
	private List<String> watchChild(String path, RegistryNotify rn) {
		try {
			CuratorWatcherAdaptor a = new CuratorWatcherAdaptor(this, path, rn);
			List<String> childList = this.curatorFramework.getChildren().usingWatcher(a).forPath(path);
			return childList;
		} catch (Exception e) {
			throw new ZookeeperException(e);
		}
	}
	
	private static class CuratorWatcherAdaptor implements CuratorWatcher{
		private String path;
		private RegistryNotify rn;
		private ZookeeperRegistryImpl registryImpl;
		CuratorWatcherAdaptor(ZookeeperRegistryImpl impl, String path, RegistryNotify rn) {
			this.registryImpl = impl;
			this.path = path;
			this.rn = rn;
		}

		@Override
		public void process(WatchedEvent event) throws Exception {
			CuratorWatcherAdaptor next = new CuratorWatcherAdaptor(this.registryImpl, path, rn);
			if (event.getType() == EventType.NodeChildrenChanged) {
				List<String> chi = registryImpl.curatorFramework.getChildren().usingWatcher(next).forPath(path);
				List<ReferenceConfig> scList = registryImpl.childList2RegistryData(chi);
				logger.info("provider changed, provider list: {}", JSON.toJSONString(chi));
				rn.notify(scList);
			}else {
				logger.info(JSON.toJSONString(event));
			}
		}
	}

	private List<ReferenceConfig> childList2RegistryData(List<String> childList){
		List<ReferenceConfig> result = Lists.newArrayList();
		for (String child : childList) {
			ReferenceConfig rd = new ReferenceConfig();
			
			String tmp = StringUtil.urlDecode(child);
			Map<String, String> properties = StringUtil.urlstring2propertis(tmp);
			ReflectUtil.appendConfig(properties, rd, InterfaceConfig.class);
			
			result.add(rd);
		}
		
		return result;
	}
	
	/**
	 * 1. 在consume节点下面创建临时节点
	 * 2. watch provider下面的子节点
	 */
	@Override
	public List<ReferenceConfig> subscribe(ReferenceConfig referenceConfig, RegistryNotify rn) {
		
		String path = this.registerData2String(String.format(CONSUME_PATH_FORMAT, referenceConfig.getInterfaceClass().getName()), referenceConfig);
		this.createEphemeralPath(path);
		
		List<String> childList = this.watchChild(String.format(PROVIDER_PATH_FORMAT, referenceConfig.getInterfaceClass().getName()), rn);
		return this.childList2RegistryData(childList);
	}

	@Override
	public void unsubscribe(ReferenceConfig referenceConfig) {
		// 好吧, 暂时服务器关闭, 自动释放		
	}
}
